package com.apexsoft.service;

import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.streaming.StreamingQueryException;

import java.util.Iterator;

/**
 * Created by caigx
 * Date: 2020/9/16
 * Time: 14:51
 * Description: No Description
 */
public class Consumer {
    public static void main(String[] args) {
        System.setProperty("java.security.auth.login.config","/home/apex/jaas.conf");
        System.setProperty("sun.security.jgss.debug","true");
        System.setProperty("javax.security.auth.useSubjectCredsOnly","false");
        System.setProperty("java.security.krb5.conf", "/home/apex/krb5.conf");

        SparkSession spark = SparkSession.builder().appName("apex_ids_real_cal")
               // .enableHiveSupport()
                .master("local[*]")
                .getOrCreate();
        spark.conf().set("spark.sql.streaming.metricsEnabled", "true");
        Dataset<String> kafka = spark.readStream()
                .format("kafka")
                .option("kafka.bootstrap.servers",  "datanode02.apex.com:9092")
                .option("kafka.security.protocol", "SASL_PLAINTEXT")
                .option("kafka.sasl.kerberos.service.name","kafka")
                .option("subscribe", "test")
                .load()
                .selectExpr("CAST(value AS STRING)")
                .as(Encoders.STRING());;
        Dataset<String> map = kafka.map(new MapFunction<String, String>() {
            @Override
            public String call(String row) throws Exception {
                System.out.println("收到--->" + row);
                return row;
            }
        }, Encoders.STRING());
        StreamingQuery console = map.writeStream().format("console").start();

        try {
            console.awaitTermination();
        } catch (StreamingQueryException e) {
            e.printStackTrace();
        }
        /*kafka.show();
        kafka.foreach(k->{
            System.out.println(k);
            System.out.println(k.get(0));
            System.out.println(k.length());
        });
        kafka.printSchema();*/
    }
}
